package com.dituhui.gps.local;

import java.text.ParseException;
import java.util.Date;
import java.util.List;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.log4j.Logger;

import com.dituhui.gps.service.AdjustGPS;
import com.dituhui.gps.service.DeAddressMatch;
import com.dituhui.gps.service.GetPos;
import com.dituhui.gps.service.GetTrack;
import com.dituhui.gps.vo.GPSPoint;
import com.dituhui.tool.io.Constants;
import com.dituhui.tool.io.ESTool;
import com.dituhui.tool.io.KafkaTool;
import com.dituhui.tool.io.KafkaTool.ConsumerRecordHandler;
import com.dituhui.tool.io.RedisTool;
import com.dituhui.tool.util.DateUtil;
import com.dituhui.tool.util.StringUtil;

import net.sf.json.JSONObject;

/**
 * GPS数据处理。从Kafka接收数据，处理后，写入Redis和ES。
 * 
 * @author liweigu
 * 
 */
public class GPSHandler {
	private static Logger LOGGER = Logger.getLogger(GPSHandler.class);
	private static String kafkaUrl = Constants.getValue("kafkaUrl");
	private static String kafkaTopic = Constants.getValue("kafkaTopic");
	private static String ConfigFile = "gps_config.properties";
	private static String esUrl = Constants.getValue("RealtimeESUrl", ConfigFile);
	private static String redisUrl = Constants.getValue("REDIS.HOST", ConfigFile);
	private static int redisPort = Integer.parseInt(Constants.getValue("REDIS.PORT", "config.properties"));
	private static String redisPwd = Constants.getValue("REDIS.PWD", ConfigFile);
	private static String RealtimeES_IndexPrefix = Constants.getValue("RealtimeES_IndexPrefix", ConfigFile);

	public static void main(String[] args) {
		boolean autoCommit = true;
		Long beginOffset = null;
		String groupId = "gps";
		// kafkaUrl = "121.40.86.232";
		// redisUrl = "121.40.192.218";
		ConsumerRecordHandler consumerRecordHandler = new ConsumerRecordHandler() {
			@Override
			public boolean handle(List<ConsumerRecord<String, String>> records) {
				for (ConsumerRecord<String, String> record : records) {
					GPSPoint gpsPoint = consumerRecordToGPSPoint(record);
					// String groupId = getGroupIdFromConsumerRecord(record);
					if (gpsPoint != null) {
						// 获取修正坐标
						double adjustedGPSX = gpsPoint.x;
						double adjustedGPSY = gpsPoint.y;
						String speedClass = null;
						try {
							// 调用定位漂移服务
							AdjustGPS adjustGPS = new AdjustGPS();
							String result = adjustGPS.getGPS(gpsPoint.x, gpsPoint.y);
							// {"result":{"x":106.00044711292863,"y":36.99902504541951,"id":0},"lineid":"9056167","speedclass":"6"}
							if (result != null) {
								JSONObject positionJsonObject = JSONObject.fromObject(result);
								JSONObject adjustedXY = (JSONObject) positionJsonObject.get("result");
								String lineId = positionJsonObject.getString("lineid");
								// 当定位漂移没有结果时，lineId是null；定位漂移有结果才给坐标和限速赋值
								if (lineId != null) {
									speedClass = positionJsonObject.getString("speedclass");
									double d = adjustedXY.getDouble("x");
									// adjustedGPSX坐标只返回小数点后6位
									adjustedGPSX = Double.parseDouble(StringUtil.formatDouble(d, GPSPoint.CoordFormat));
									d = adjustedXY.getDouble("y");
									// adjustedGPSY坐标只返回小数点后6位
									adjustedGPSY = Double.parseDouble(StringUtil.formatDouble(d, GPSPoint.CoordFormat));
								}
							}
						} catch (Exception ex) {
							// 如果发生异常，修正坐标的值使用跟原始坐标相同的。
							String warning = "获取修正坐标发生异常。";
							LOGGER.warn(warning, ex);
						}
						gpsPoint.setReviseX(adjustedGPSX);
						gpsPoint.setReviseY(adjustedGPSY);

						// 道路限速默认值120
						int speedLimit = 120;
						if (speedClass != null) {
							String speedLimitValue = Constants.getValue("GPS.SPEEDCLASS." + speedClass, ConfigFile);
							if (speedLimitValue != null) {
								try {
									speedLimit = Integer.parseInt(speedLimitValue);
								} catch (Exception e) {
									LOGGER.debug("解析限速值失败。speedLimitValue=" + speedLimitValue);
								}
							}
						}
						gpsPoint.setSpeedLimit(speedLimit);

						// 获取位置
						DeAddressMatch deAddressMatch = new DeAddressMatch();
						String address = "";
						try {
							address = deAddressMatch.getAddress(gpsPoint.x, gpsPoint.y);
							// System.out.println("address = " + address);
						} catch (Exception ex) {
							String msg = "未获得位置。x=" + gpsPoint.x + ",y=" + gpsPoint.y;
							LOGGER.debug(msg, ex);
						}
						gpsPoint.setAddress(address);

						// 计算车辆状态
						// int carStatus = calculateCarStatus(gpsX, gpsY, velocity, speedLimit, obj_gpsDataTime, plateNumber, accountId);
						// gpsPoint.setStatus(carStatus);

						// 当天日期
						String today = DateUtil.getDayFormat().format(new Date());
						String gpsDay = DateUtil.getDayFormat().format(gpsPoint.getGPSTime());
						// 写入redis
						if (today.equals(gpsDay)) {
							// 当天的、或测试数据，才存入Redis
							RedisTool redisTool = new RedisTool();

							// 转换为字符串
							String gpsData = gpsPoint.toCSVString();

							// 放入实时位置
							String redisKey = GetPos.getRedisKey(gpsPoint.getAccountId(), gpsPoint.getGroupId());
							redisTool.setString(redisKey, gpsData, redisUrl, redisPort, redisPwd);

							// 放入当日轨迹
							redisKey = GetTrack.getRedisKey(today, gpsPoint.getAccountId());
							redisTool.appendString(redisKey, gpsData, redisUrl, redisPort, redisPwd);
						}

						// 写入ES
						String index = RealtimeES_IndexPrefix + today;
						String[] columnNames;
						String[] columnValues;
						// if (groupId == null) {
						columnNames = new String[] { "aid", "plateNumber", "gps" };
						columnValues = new String[] { gpsPoint.getAccountId(), gpsPoint.getPlateNumber(), gpsPoint.toCSVString() };
						// } else {
						// columnNames = new String[] { "aid", "plateNumber", "gps", "groupId" };
						// columnValues = new String[] { gpsPoint.getAccountId(), gpsPoint.getPlateNumber(), gpsPoint.toCSVString(), groupId };
						// }
						ESTool.write(index, columnNames, columnValues, esUrl);
					}
				}
				return true;
			}
		};
		LOGGER.info("KafkaTool.consume");
		KafkaTool.consume(kafkaUrl, kafkaTopic, consumerRecordHandler, groupId, autoCommit, beginOffset);
	}

	// private static String getGroupIdFromConsumerRecord(ConsumerRecord<String, String> record) {
	// String groupId = null;
	//
	// if (record != null) {
	// String accountId = record.key();
	// String value = record.value();
	// if (accountId != null && value != null && value.indexOf(StringUtil.AngularBracket) != -1) {
	// // 104.066541^30.572269^-1^2016-12-20 14:08:36^groupId^phone
	// String[] arr = value.split(StringUtil.AngularBracketForSplit);
	// if (arr.length >= 5) {
	// String s = arr[4];
	// if (s.length() > 0 && !s.equals(StringUtil.NULL)) {
	// groupId = s;
	// }
	// }
	// }
	// } else {
	// // 参数非法
	// }
	//
	// return groupId;
	// }

	private static GPSPoint consumerRecordToGPSPoint(ConsumerRecord<String, String> record) {
		GPSPoint gpsPoint = null;

		if (record != null) {
			String accountId = record.key();
			String value = record.value();
			if (accountId != null && value != null && value.indexOf(StringUtil.AngularBracket) != -1) {
				// 104.066541^30.572269^-1^2016-12-20 14:08:36^groupId^phone^speed
				String[] arr = value.split(StringUtil.AngularBracketForSplit);
				double x = Double.parseDouble(arr[0]);
				double y = Double.parseDouble(arr[1]);
				int accuracy = 0;
				try {
					accuracy = Integer.parseInt(arr[2]);
				} catch (NumberFormatException e1) {
					LOGGER.warn("value=" + value);
				}
				Date gpsTime = null;
				try {
					gpsTime = DateUtil.getGPSTimeFormat().parse(arr[3]);

					gpsPoint = new GPSPoint(x, y);
					gpsPoint.setAccountId(accountId);
					gpsPoint.setGPSTime(gpsTime);
					gpsPoint.setAccuracy(accuracy);

					if (arr.length >= 6) {
						String groupId = arr[4];
						// UNDEFINED和NULL是非法值
						if (!StringUtil.UNDEFINED.equalsIgnoreCase(groupId) && !StringUtil.NULL.equalsIgnoreCase(groupId)) {
							gpsPoint.setGroupId(groupId);
						}
						String phone = arr[5];
						if (!StringUtil.UNDEFINED.equalsIgnoreCase(groupId) && !StringUtil.NULL.equalsIgnoreCase(groupId)) {
							gpsPoint.setLoginMobile(phone);
						}
					}
					if (arr.length >= 7) {
						String speed = arr[6];
						if (!StringUtil.UNDEFINED.equalsIgnoreCase(speed) && !StringUtil.NULL.equalsIgnoreCase(speed)) {
							gpsPoint.setSpeed(Double.parseDouble(speed));
						}
					}
				} catch (ParseException e) {
					e.printStackTrace();
				}
			} else {
				// 参数非法
			}
		}

		return gpsPoint;
	}

	// private int calculateCarStatus(String gpsX, String gpsY, String velocity, double speedLimit, Date gpsTime, String plateNumber, String accountId) {
	// // 默认设置为正常行驶
	// int carStatus = TracksConstants.TRUCK_STATUS_ONLINE_RUNNING;
	//
	// try {
	// GPSPoint gpsPoint = new GPSPoint(Double.parseDouble(gpsX), Double.parseDouble(gpsY));
	// double speed = -1;
	// try {
	// speed = Double.parseDouble(velocity);
	// } catch (NumberFormatException e) {
	// // TODO Auto-generated catch block
	// }
	// gpsPoint.setSpeed(speed);
	// gpsPoint.setSpeedLimit(speedLimit);
	// gpsPoint.setPlateNumber(plateNumber);
	// gpsPoint.setAccountId(accountId);
	// gpsPoint.setGPSTime(gpsTime);
	// // 添加到队列
	// CaculateCarStatusTool.addToQueue(gpsPoint);
	// // 根据速度计算行驶情况
	// CaculateCarStatusTool.caculateSpeedStatus(gpsPoint);
	// if (gpsPoint.getStatus() != TracksConstants.TRUCK_STATUS_UNKNOW) {
	// // 如果根据速度计算改变了Status就保存
	// carStatus = gpsPoint.getStatus();
	// } else {
	// // 不会进入此分支代码
	// // // 计算停留情况
	// // CaculateCarStatusTool.caculateStopStatus(gpsPoint);
	// // if (gpsPoint.getStatus() != TracksConstants.TRUCK_STATUS_UNKNOW) {
	// // carStatus = gpsPoint.getStatus();
	// // }
	// }
	// } catch (NumberFormatException e) {
	// // 正常情况不会发生异常。
	// String message = "AccessDBThread.calculateCarStatus发生解析坐标异常。";
	// LOG.debug(message, e);
	// }
	//
	// return carStatus;
	// }
}
